package jdbcState;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.trident.TridentTopology;
import org.apache.storm.trident.state.StateType;
import org.apache.storm.trident.testing.FixedBatchSpout;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;
import org.apache.storm.utils.Utils;

/**
 * 事物Trident-MySQL Topology
 * @author mengyao
 *
 */
public class JDBCTopology {

    public static void main(String[] args) throws Exception {
        TridentTopology topology = new TridentTopology();

        //Spout数据源
        FixedBatchSpout spout = new FixedBatchSpout(
                new Fields("tels"),
                2,
                new Values("189111	3"),
                new Values("135111	7"),
                new Values("189111	2"),
                new Values("158111	5"),
                new Values("159111	6"),
                new Values("159111	3"),
                new Values("158111	5")
        );
        spout.setCycle(false);

        // 为将状态持久化到MySQL数据库做准备
        //State持久化配置属性
        JDBCStateConfig config = new JDBCStateConfig();
        config.setDriver("com.mysql.jdbc.Driver");
        config.setUrl("jdbc:mysql://niit-master:3306/sem7");
        config.setUsername("root");
        config.setPassword("root");
        // 设置数据库
        // CREATE TABLE `tbl_tel` ( `id` INT NOT NULL AUTO_INCREMENT , `sum` INT NOT NULL , `txid` INT NOT NULL , `time` VARCHAR(50) NOT NULL , `tel` VARCHAR(50) NOT NULL , PRIMARY KEY (`id`)) ENGINE = InnoDB;


        config.setName("name");
        config.setLng("lng");
        config.setTable("tbl_tel");


        config.setBatchSize(10);
        config.setCacheSize(10);
        config.setType(StateType.TRANSACTIONAL);

        topology.newStream("spout", spout)
                // 拆分字段，结果类似于： 189111	3,189111,3
                .each(new Fields("tels"), new KeyValueFun(), new Fields("tel", "money"))
                // 按照tel对在一个批次内的所有的tuple进行分组，相同tel的tuple会被执行下面的SumCombinerAgg聚合操作
                .groupBy(new Fields("tel"))
                // 对于持久化聚合，第一个参数必须是一个State的Factory类
                //              第三个参数必须是一个聚合类
                .persistentAggregate(JDBCState.getFactory(config),
                        new Fields("money"),
                        new SumCombinerAgg(),
                        new Fields("sum"));

        LocalCluster cluster = new LocalCluster();
        cluster.submitTopology("test1", new Config(), topology.build());

        Utils.sleep(20*1000);
        cluster.shutdown();
        System.exit(0);
    }

}





